package we.dispatcher;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import org.json.JSONObject;
import we.common.CCommonHandler;
import we.config.CConstant;

public class CDispatcherHandler implements Runnable {

    private static CDispatcherHandler _instance;
    private static final Object lock = new Object();
    private Executor writer;
    private Executor reader;
    private BlockingQueue<SelectionKey> in;
    private BlockingQueue<CMessage> out;
    private Selector selector;
    private ServerSocketChannel serverChannel;
    private SelectionKey serverkey;

    private CDispatcherHandler() {
        in = new LinkedBlockingQueue<SelectionKey>();
        out = new LinkedBlockingQueue<CMessage>();
        reader = Executors.newFixedThreadPool(CConstant.MAX_INQUEUE_WORKER);
        writer = Executors.newFixedThreadPool(CConstant.MAX_OUTQUEUE_WORKER);

        for (int i = 0; i < CConstant.MAX_INQUEUE_WORKER; i++) {
            reader.execute(new CReadHandler(in));
        }

        for (int i = 0; i < CConstant.MAX_OUTQUEUE_WORKER; i++) {
            writer.execute(new CWriteHandler(out));
        }

        try {
            selector = Selector.open();
            serverChannel = ServerSocketChannel.open();
            serverChannel.socket().bind(new InetSocketAddress(CConstant.SERVER_PORT));
            serverChannel.configureBlocking(false);
            serverkey =
                    serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            // TODO
            CCommonHandler.writeErrLog(e);
            System.out.println("Server is stoped");
            System.exit(1);
        }
        System.out.println("Server is running");
    }

    public static CDispatcherHandler instance() {
        if (_instance == null) {
            synchronized (lock) {
                if (_instance == null) {
                    _instance = new CDispatcherHandler();
                }
            }
        }
        return _instance;
    }

    @Override
    public void run() {
        CCommonHandler.trace("run CDispathcer Hanler");
        while (true) {
            try {
                selector.select();
                Set keys = selector.selectedKeys();
                for (Iterator i = keys.iterator(); i.hasNext();) {
                    SelectionKey key = (SelectionKey) i.next();
                    i.remove();
                    if (key == serverkey) {
                        if (key.isAcceptable()) {
                            SocketChannel client = serverChannel.accept();
                            client.configureBlocking(false);
                            client.register(selector, SelectionKey.OP_READ);
                            CCommonHandler.trace("acceptsss");                    
                        }
                    } else {
                        if (key.isValid()) {
                            Boolean flag = (Boolean) key.attachment();
                            if (flag == null || !flag) {
                                key.attach(true);
                                in.offer(key);
                            }
                        }
                    }
                }
            } catch (IOException e) {
                CCommonHandler.writeErrLog(e);
            }
        }
    }

    public void sendResponse(SocketChannel channel, JSONObject json) {
        StringBuilder sb = new StringBuilder(json.toString());
        sb.append("\n");

        CMessage msg = new CMessage();
        msg.channel = channel;
        msg.data = sb.toString();
        out.offer(msg);
    }

    public void sendResponse(List<SocketChannel> channelList,
            JSONObject json) {
        StringBuilder sb = new StringBuilder(json.toString());
        sb.append("\n");
        String data = sb.toString();
        for (SocketChannel channel : channelList) {
            CMessage msg = new CMessage();
            msg.channel = channel;
            msg.data = data;
            out.offer(msg);
        }
    }

    public void sendPolicy(SocketChannel channel) {
        CMessage msg = new CMessage();
        msg.channel = channel;
        msg.policyResponse = true;
        out.offer(msg);
    }
}
